package org.joy.hpc.analyzer.examples.sogou;

import java.io.ByteArrayOutputStream;
import java.io.EOFException;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;

import org.apache.hadoop.conf.Configuration;

/**
 * 从输入流中读取Sogou互联网文档
 * 
 * @author Jeremy Chow(coderplay@gmail.com), Song Liulamfeeling@126.com
 */
public class SogouCorpusReader {

	/* the default buffer size - should not be too large */
	private static final int DEFAULT_BUFFER_SIZE = 1024 * 4;
	private int bufferSize = DEFAULT_BUFFER_SIZE;

	/* input stream which we will get documents from */
	private InputStream in;
	/* a buffer stores temporary bytes readed from input stream */
	private byte[] buffer;
	/* the number of bytes of real data in the buffer */
	private int bufferLength = 0;
	/* the current position of the buffer */
	private int posAtBuffer = 0;
	/* the buffer position of input stream */
	private long bufferPos = 0;
	/* current document position of input stream */
	private long currentDocPosn = 0;
	/* the end pos of last document */
	private int preDocEnd = 0;
	/* start tag */
	private final static byte[] START_TAG = "<DOC>".getBytes();
	/* end tag */
	private final static byte[] END_TAG = "</DOC>".getBytes();

	/**
	 * create instance with inputstream, using default buffer size
	 * 
	 * @param in
	 *            input stream for sougou corpus reader
	 * @throws IOException
	 */
	public SogouCorpusReader(InputStream in) throws IOException {
		this(in, DEFAULT_BUFFER_SIZE);
	}

	/**	 * create a instance with given inputstream and buffersize
	 * 
	 * @param in
	 *            given input stream
	 * @param bufferSize
	 *            the buffersize, SHOULD NOT BE SMALLER THAN 4*1024
	 * @throws IOException
	 */
	public SogouCorpusReader(InputStream in, int bufferSize) throws IOException {
		this.in = in;
		this.bufferSize = bufferSize;
		this.buffer = new byte[this.bufferSize];
	}

	/**
	 * create an instance with the parameters given in configuration
	 * 
	 * @param in
	 *            inputstream used in our instance
	 * @param conf
	 *            the configuration used to read parameters.
	 * @throws IOException
	 */
	public SogouCorpusReader(InputStream in, Configuration conf)
			throws IOException {
		this(in, conf.getInt("redpoll.sogou.doc.buffersize",
				DEFAULT_BUFFER_SIZE));
	}

	/**
	 * Gets a {@link DocumentInput.text.Document} instance from sogou text file.
	 * If it reached EOF, it willreturn the end position of inputstream.
	 * 
	 * @param doc
	 *            a {@link DocumentInput.text.Document} instance getting from
	 *            sogou text file.
	 * @return the end position of this document, -1 if it reached EOF.
	 * @throws IOException
	 */
	public long nextDoc(SogouDocumentInput doc) throws IOException {
		try {
			// find <doc> tag
			int index, start;
			do {
				adjustBuffer();

				index = findBytes(buffer, bufferLength, START_TAG, preDocEnd,
						true);

				if (index == -1) {
					posAtBuffer = bufferLength;
				} else {
					posAtBuffer = index;
					start = index;
					currentDocPosn = bufferPos + posAtBuffer;
					break;
				}
			} while (true);

			int pos = 0;
			// our bytes buffer
			ByteArrayOutputStream baos = new ByteArrayOutputStream(bufferLength);
			// find </doc> tag
			do {
				adjustBuffer();

				index = findBytes(buffer, bufferLength, END_TAG, start, true);

				if (index == -1) {
					// the </doc> tag exceed the end of current buffer, we must
					// save the first part of our doc
					baos.write(buffer, posAtBuffer, bufferLength - posAtBuffer);

					start = 0;
					pos += bufferLength - posAtBuffer;
					posAtBuffer = bufferLength;
				} else {
					// read the bytes between the <doc> and </doc>
					preDocEnd = index;
					posAtBuffer = index;
					baos.write(buffer, start, posAtBuffer - start);

					doc.setRawBytes(baos.toByteArray());

					return bufferPos + posAtBuffer + END_TAG.length;
				}
			} while (true);
		} catch (EOFException ex) {
			return -1;
		}
	}

	public long getPosition() {
		return bufferPos + posAtBuffer;
	}

	/**
	 * Close the underlying stream.
	 * 
	 * @throws IOException
	 */
	public void close() throws IOException {
		in.close();
	}

	private void adjustBuffer() throws IOException, EOFException {
		if (posAtBuffer >= bufferLength) {
			preDocEnd = 0;
			bufferPos += bufferLength;
			posAtBuffer = 0;
			bufferLength = in.read(buffer, 0, DEFAULT_BUFFER_SIZE
					- END_TAG.length + 1);
			if (bufferLength <= 0) {
				posAtBuffer = -1;
				throw new EOFException();
			}
			int i = bufferLength - 1;
			// don't truncate <DOC> OR </DOC> tags
			while (i >= 0 && i > bufferLength - END_TAG.length) {
				// if we see any closed tags first, it seems all the tags has
				// been buffered completely.
				if (buffer[i] == '>')
					break;
				// if there are any tag has been started , but not closed yet
				if (buffer[i] == '<') {
					// try to read to >
					int j = bufferLength;
					while (j < DEFAULT_BUFFER_SIZE) {
						int t = in.read();
						// if we already reached the end of file
						if (t == -1)
							throw new EOFException();
						buffer[j] = (byte) t;
						bufferLength++;
						if (buffer[j] == '>') {
							break;
						}
						j++;
					}
					break;
				}
				i--;
			}
		}
	}

	/**
	 * find target bytes in given source
	 * 
	 * @param src
	 *            the source bytes
	 * @param target
	 *            the target bytes
	 * @param index
	 *            the start index you want to search
	 * @param caseSensitive
	 *            whether the search is case sensitive
	 * @return the position that the target first appear in source
	 */
	public static int findBytes(byte[] src, byte[] target, int index,
			boolean caseSensitive) {
		return findBytes(src, src.length, target, index, caseSensitive);
	}

	public static int findBytes(byte[] src, int range, byte[] target,
			int index, boolean caseSensitive) {
		index = index < 0 ? 0 : index;
		for (int i = index; i < range; i++) {
			int j = 0;
			boolean match = true;
			while (j < target.length) {
				byte alternative = target[j];
				if (!caseSensitive) {
					if (alternative > 97) {
						alternative -= 32;
					} else {
						alternative += 32;
					}
				}
				if (i + j == range
						|| (src[i + j] != target[j] && src[i + j] != alternative)) {
					match = false;
					break;
				}
				j++;
			}
			if (match) {
				return i;
			}
		}
		return -1;
	}

	/**
	 * for test
	 */
	public static void main(String[] args) throws IOException {

		FileInputStream in = new FileInputStream(args[0]);
		// in.skip(10);
		SogouCorpusReader reader = new SogouCorpusReader(in);
		SogouDocumentInput doc = new SogouDocumentInput();
		long pos;
		long start = System.currentTimeMillis();
		int i = 0;

		while ((pos = reader.nextDoc(doc)) >= 0) {
			System.out.println("docURL:" + doc.getURL());
			// System.out.println("docID:" + doc.getDocumentId());
			i++;
		}
		System.out.println("in total:" + i);
		System.out.println(System.currentTimeMillis() - start);
		in.close();
	}
}
